/*
 * Copyright (c) 2008, Solido Systems
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * Redistributions of source code must retain the above copyright notice, this
 * list of conditions and the following disclaimer.
 *
 * Redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the following disclaimer in the documentation
 * and/or other materials provided with the distribution.
 *
 * Neither the name of Solido Systems nor the names of its contributors may be
 * used to endorse or promote products derived from this software without
 * specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

package com.solidosystems.objectkitchen.storage;

import java.io.*;
import java.util.*;
import com.solidosystems.objectkitchen.data.*;
import com.solidosystems.objectkitchen.index.*;

public class Block implements Storage{
	int[] primes={	  2,	3,		5,		7,		11,		13,		17,		19,
					 23,	29,		31,		37,		41,		43,		47,		53,
					 59, 	61,		67,		71,		73,		79,		83,		89,
					 97,	101,	103,	107,	109,	113,	127,	131,
					137,    139,    149,    151,    157,    163,    167,    173, 
		   			179,    181,    191,    193,    197,    199,    211,    223,
		    		227,    229, 	233,    239,    241,    251,    257,    263,
					269,    271,    277,    281,	283,    293,    307,    311};
	int[] lookup;
	int seed=1;
	
	private String location;
	private NodeManager idIndex;
	private NodeManager groupIndex;
	
	private Map<Integer,String> idfiles;
	private Map<Integer,String> groupfiles;

	private int currentIdFile=-1;
	private int currentGroupFile=-1;
	
	private long CNT_GETDOC=0;
	private long CNT_GETORCREATEDOC=0;
	private long CNT_PUTDOC=0;
	private long CNT_DELDOC=0;
	private long CNT_HASDOC=0;
	private long CNT_GETATT=0;
	private long CNT_PUTATT=0;
	private long CNT_DELATT=0;
	
	public synchronized Map<String,Long> getCounters(){
		HashMap<String,Long> map=new HashMap<String,Long>();
		map.put("CNT_GETDOC",CNT_GETDOC);
		map.put("CNT_GETORCREATEDOC",CNT_GETORCREATEDOC);
		map.put("CNT_PUTDOC",CNT_PUTDOC);
		map.put("CNT_DELDOC",CNT_DELDOC);
		map.put("CNT_HASDOC",CNT_HASDOC);
		map.put("CNT_GETATT",CNT_GETATT);
		map.put("CNT_PUTATT",CNT_PUTATT);
		map.put("CNT_DELATT",CNT_DELATT);
		return map;
	}
	
	public int getHash(String str){
		int result=seed;
		
		try{
			int num=0;
			for(byte b:str.getBytes("UTF-8")){
				result=result*lookup[(b+num)&4095]+primes[(b*num)&63];
				num++;
			}
		}catch(Exception e){
			e.printStackTrace();
		}
		return result;
	}
	
	
	public Block(String location, int seed) throws IOException{
		this.seed=seed;
		lookup=new int[4096];
		int adder=1337;
		for(int i=0;i<4096;i++){
			adder=i*adder+primes[i%63]+i;
			lookup[i]=adder%1024-512;
		}
		this.seed=lookup[seed&4095];

		idfiles=new HashMap<Integer,String>();
		groupfiles=new HashMap<Integer,String>();
		
		File ftest=new File(location);
		if(!ftest.exists()){
			ftest.mkdir();
		}
		if(!location.endsWith(File.separator))location=location+File.separator;
		this.location=location;
		idIndex=new NodeManager(location+"id.index");
		long smallestid=-1;
		long smallestgroup=-1;
		String[] list=ftest.list();
		for(String filename:list){
			if(filename.endsWith(".data")&&filename.startsWith("slice_")){
				String snum=filename.substring(0,filename.indexOf(".data")).substring(6);
				try{
					int inum=Integer.parseInt(snum);
					idfiles.put(inum,location+filename);
					ftest=new File(location+filename);
					if(smallestid==-1){
						smallestid=ftest.length();
						currentIdFile=inum;
					}else{
						if(ftest.length()<smallestid){
							smallestid=ftest.length();
							currentIdFile=inum;
						}
					}
				}catch(Exception e){
					e.printStackTrace();
				}
			}else if(filename.endsWith(".group")&&filename.startsWith("slice_")){
				String snum=filename.substring(0,filename.indexOf(".group")).substring(6);
				try{
					int inum=Integer.parseInt(snum);
					idfiles.put(inum,location+filename);
					ftest=new File(location+filename);
					if(smallestgroup==-1){
						smallestgroup=ftest.length();
						currentGroupFile=inum;
					}else{
						if(ftest.length()<smallestgroup){
							smallestgroup=ftest.length();
							currentGroupFile=inum;
						}
					}
				}catch(Exception e){
					e.printStackTrace();
				}
			}
		}
		if(currentIdFile==-1){
			// No data files yet
			currentIdFile=0;
			idfiles.put(0,getFilename(currentIdFile));
		}
		if(currentGroupFile==-1){
			// No data files yet
			currentGroupFile=0;
			idfiles.put(0,getFilename(currentGroupFile));
		}
	}
	
	private int getCurrentIdFileNumber() throws IOException{
		String filename=idfiles.get(currentIdFile);
		File ftest=new File(filename);
		if(ftest.exists()){
			if(ftest.length()>256l*1024l*1024l){
				while(idfiles.containsKey(currentIdFile))currentIdFile++;
				filename=getFilename(currentIdFile);
				idfiles.put(currentIdFile,filename);
			}
		}
		return currentIdFile;
	}
	
	private int getCurrentGroupFileNumber() throws IOException{
		String filename=groupfiles.get(currentIdFile);
		File ftest=new File(filename);
		if(ftest.exists()){
			if(ftest.length()>256l*1024l*1024l){
				while(groupfiles.containsKey(currentGroupFile))currentGroupFile++;
				filename=getFilename(currentGroupFile);
				groupfiles.put(currentGroupFile,filename);
			}
		}
		return currentGroupFile;
	}
	
	private String getFilename(int num){
		return location+"slice_"+num+".data";
	}
	
	private Document readDocument(int fileId,long location) throws IOException{
		DataInputStream buf=new DataInputStream(new BufferedInputStream(new FileInputStream(idfiles.get(fileId))));
		long position=0;
		while(position<location)position+=buf.skipBytes((int)(location-position));
		Document doc=Document.readFromStream(buf);
		buf.close();
		return doc;
	}
	
	private Document readDocument(IndexEntry entry) throws IOException{
		return readDocument(entry.getFileId(),entry.getFileOffset());
	}
	
	public synchronized Document getDocument(String id, int version) throws IOException{
		CNT_GETDOC++;
		IndexEntry entry=idIndex.findEntry(id,getHash(id));
		if(entry!=null){
			if(idfiles.containsKey(entry.getFileId())){
				Document doc=readDocument(entry.getFileId(),entry.getFileOffset());
				if(doc!=null){
					if(doc.getVersion()==version)return doc;
					List<Long> history=doc.getHistory();
					for(long location:history){
						doc=readDocument(entry.getFileId(),entry.getFileOffset());
						if(doc!=null){
							if(doc.getVersion()==version)return doc;
						}
					}
				}
			}
		}
		return null;
	}
	
	public synchronized void deleteDocument(String id) throws IOException{
		CNT_DELDOC++;
		IndexEntry entry=idIndex.findEntry(id,getHash(id));
		if(entry!=null){
			entry.setDeleted(true);
			idIndex.updateIndexEntry(entry);
		}
	}
	
	public synchronized boolean hasDocument(String id) throws IOException{
		CNT_HASDOC++;
		IndexEntry entry=idIndex.findEntry(id,getHash(id));
		if(entry!=null){
			if(!entry.isDeleted()){
				return true;
			}
		}
		return false;
	}
	
	public synchronized Document getDocument(String id) throws IOException{
		CNT_GETDOC++;
		IndexEntry entry=idIndex.findEntry(id,getHash(id));
		if(entry!=null){
			if(!entry.isDeleted()){
				if(idfiles.containsKey(entry.getFileId())){
					Document doc=readDocument(entry.getFileId(),entry.getFileOffset());
					return doc;
				}
			}
		}
		return null;
	}
	
	public synchronized void deleteAttachment(String id,String title) throws IOException{
		CNT_DELATT++;
		Document doc=getDocument(id);
		if(doc==null)return;
		deleteAttachment(doc,title);
	}
	
	public synchronized void deleteAttachment(Document doc,String title) throws IOException{
		CNT_DELATT++;
		doc.removeAttachment(title);
		putDocument(doc);
	}
	
	public byte[] getAttachment(String id,String title) throws IOException{
		CNT_GETATT++;
		return getAttachment(getDocument(id),title);
	}
	
	public synchronized byte[] getAttachment(Document doc,String title) throws IOException{
		CNT_GETATT++;
		if(doc==null)return null;
		if(!doc.hasAttachment(title))return null;
		IndexEntry entry=idIndex.findEntry(doc.getId(),getHash(doc.getId()));
		if(entry!=null){
			long location=doc.getAttachmentOffset(title);
			DataInputStream buf=new DataInputStream(new BufferedInputStream(new FileInputStream(idfiles.get(entry.getFileId()))));
			long position=0;
			while(position<location)position+=buf.skipBytes((int)(location-position));
			int size=buf.readInt();
			byte[] result=new byte[size];
			buf.readFully(result);
			buf.close();
			return result;
		}
		return null;
	}
	
	public synchronized void putAttachment(String id,String title,byte[] data) throws IOException{
		CNT_PUTATT++;
		Document doc=getDocument(id);
		if(doc==null)return;
		putAttachment(doc,title,data);
	}
	
	public synchronized void putAttachment(Document doc,String title,byte[] data) throws IOException{
		CNT_PUTATT++;
		IndexEntry entry=idIndex.findEntry(doc.getId(),getHash(doc.getId()));
		if(entry!=null){
			String filename=getFilename(entry.getFileId());
			File ftest=new File(filename);
			long offset=0;
			if(ftest.exists())offset=ftest.length();
			DataOutputStream buf=new DataOutputStream(new BufferedOutputStream(new FileOutputStream(filename,true)));
			buf.writeInt(data.length);
			buf.write(data,0,data.length);
			buf.flush();
			buf.close();
			doc.setAttachment(title,offset);
			putDocument(doc);
		}
	}
	
	public synchronized Document getOrCreateDocument(String id) throws IOException{
		CNT_GETORCREATEDOC++;
		Document doc=getDocument(id);
		if(doc!=null)return doc;
		return new Document(id);
	}
	
	public synchronized void addToGroupIndex(String group, String id) throws IOException{
		IndexEntry entry=groupIndex.findEntry(group,getHash(group));
		if(entry==null){
			// No previous versions
			int filenum=getCurrentGroupFileNumber();
			String filename=getFilename(filenum);
			File ftest=new File(filename);
			long offset=0;
			if(ftest.exists())offset=ftest.length();
			DataOutputStream buf=new DataOutputStream(new BufferedOutputStream(new FileOutputStream(filename,true)));
			// write out the new group list file
			GroupPage page=new GroupPage(group,8192);
			page.add(id);
			page.writeToFile(buf);
			buf.flush();
			buf.close();
			groupIndex.addIndexEntry(group,getHash(group),filenum,offset);
		}else{
			String filename=getFilename(entry.getFileId());
			DataInputStream buf=new DataInputStream(new BufferedInputStream(new FileInputStream(filename)));
		}
	}
	
	public synchronized void addToGroupIndex(String id) throws IOException{
		StringTokenizer strtkn=new StringTokenizer(id,"/");
		String group="/";
		while(strtkn.hasMoreTokens()){
			String str=strtkn.nextToken();
			if(!id.endsWith(str)){
				group=group+str+"/";
				addToGroupIndex(group,id);
			}
		}
	}
	
	public synchronized void putDocument(Document doc) throws IOException{
		CNT_PUTDOC++;
		IndexEntry entry=idIndex.findEntry(doc.getId(),getHash(doc.getId()));
		if(entry==null){
			// No previous versions
			int filenum=getCurrentIdFileNumber();
			String filename=getFilename(filenum);
			File ftest=new File(filename);
			long offset=0;
			if(ftest.exists())offset=ftest.length();
			DataOutputStream buf=new DataOutputStream(new BufferedOutputStream(new FileOutputStream(filename,true)));
			doc.writeToStream(buf);
			buf.flush();
			buf.close();
			idIndex.addIndexEntry(doc.getId(),getHash(doc.getId()),filenum,offset);
			addToGroupIndex(doc.getId());
		}else{
			if(entry.isDeleted()){
				// New document over previously used id
				String filename=getFilename(entry.getFileId());
				File ftest=new File(filename);
				long offset=ftest.length();
				DataOutputStream buf=new DataOutputStream(new BufferedOutputStream(new FileOutputStream(filename,true)));
				doc.writeToStream(buf);
				buf.flush();
				buf.close();
				entry.setFileOffset(offset);
				entry.setDeleted(false);
				idIndex.updateIndexEntry(entry);
			}else{
				// New version of existing document
				Document last=getDocument(doc.getId());
				List<Long> history=last.getHistory();
				history.add(0,entry.getFileOffset());
				doc.setHistory(history);
				doc.incrementVersion();
				String filename=getFilename(entry.getFileId());
				File ftest=new File(filename);
				long offset=ftest.length();
				DataOutputStream buf=new DataOutputStream(new BufferedOutputStream(new FileOutputStream(filename,true)));
				doc.writeToStream(buf);
				buf.flush();
				buf.close();
				entry.setFileOffset(offset);
				idIndex.updateIndexEntry(entry);
			}
		}
	}
}